package edu.tyut.game;

import edu.tyut.pojo.FlyingFlowerOrderData;
import edu.tyut.pojo.PoemRecommend;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.api.java.tuple.Tuple6;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

public class GameStream {
    public static void main(String[] args) throws Exception {

        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

        // 2. 读取数据源 TODO
        DataStreamSource<String> lineDS = env.socketTextStream("centos01", 7777);

        SingleOutputStreamOperator<Tuple5<String, Integer, Integer, Integer, Long>> operatorFlyGame = lineDS.filter(it -> it.contains("flyGame"))
                .map(it -> {
                    String[] str = it.split(",");
                    return new Tuple5<>(str[0], Integer.parseInt(str[1]), Integer.parseInt(str[2]), Integer.parseInt(str[3]), Long.parseLong(str[4]));
                }).returns(Types.TUPLE(Types.STRING, Types.INT, Types.INT, Types.INT, Types.LONG));

        SingleOutputStreamOperator<Tuple6<String, Integer, Integer, Integer, Integer, Long>> operatorPoemRecommend = lineDS.filter(it -> it.contains("PoemRecommend"))
                .map(it -> {
                    String[] str = it.split(",");
                    return new Tuple6<>(str[0], Integer.parseInt(str[1]), Integer.parseInt(str[2]), Integer.parseInt(str[3]), Integer.parseInt(str[4]), Long.parseLong(str[5]));
                }).returns(Types.TUPLE(Types.STRING, Types.INT, Types.INT, Types.INT, Types.INT, Types.LONG));

        // 3. 创建Schema
        final Schema schemaFlyGame = Schema.newBuilder()
                .column("f0", DataTypes.STRING())
                .column("f1", DataTypes.INT())
                .column("f2", DataTypes.INT())
                .column("f3", DataTypes.INT())
                .column("f4", DataTypes.BIGINT())
                .build();

        final Schema schemaPoemRecommend = Schema.newBuilder()
                .column("f0", DataTypes.STRING())
                .column("f1", DataTypes.INT())
                .column("f2", DataTypes.INT())
                .column("f3", DataTypes.INT())
                .column("f4", DataTypes.INT())
                .column("f5", DataTypes.BIGINT())
                .build();

        // 4. 临时表
        String flyGameView = "flyGameView";
        String poemRecommendView = "poemRecommendView";

        tEnv.createTemporaryView(flyGameView, operatorFlyGame, schemaFlyGame);
        tEnv.createTemporaryView(poemRecommendView, operatorPoemRecommend, schemaPoemRecommend);

        // 5. 创建表
        Table flyGameTable = tEnv.sqlQuery("select f0 as flyGame, f1 as uid, f2 as total_rounds, f3 as win_rounds, f4 as time_stamp from flyGameView");
        // flyGameTable.execute().print();
        Table flyGameTableResult = flyGameTable
                .groupBy($("flyGame"), $("uid"))
                // , String.valueOf($("输赢(0,1)").dividedBy($("总局数")).as("胜率"))
                .select($("flyGame"), $("uid"), $("total_rounds").sum().as("total_rounds"), $("win_rounds").sum().as("win_rounds"))
                .select($("flyGame"), $("uid"), $("total_rounds"), $("win_rounds"),
                        $("win_rounds").cast(DataTypes.DOUBLE()).dividedBy($("total_rounds").cast(DataTypes.DOUBLE())).as("win_rate"),
                        $("total_rounds").minus($("win_rounds")).as("lose_rounds"))
                .select($("uid"), $("total_rounds"), $("win_rounds"), $("win_rate"))
                .distinct();

        Table poemRecommendViewTable = tEnv.sqlQuery("select f0 as PoemRecommend, f1 as uid, f2 as poemId, f3 as poetId, f4 as click_count, f5 as time_stamp from poemRecommendView");
        // poemRecommendViewTable.execute().print();
        Table poemRecommendResult = poemRecommendViewTable
                .groupBy($("PoemRecommend"), $("uid"), $("poemId"))
                .select($("PoemRecommend"), $("uid"), $("poemId"), $("click_count").sum().as("click_count"))
                .select($("uid"), $("poemId"), $("click_count"))
                .distinct();

        SingleOutputStreamOperator<FlyingFlowerOrderData> flyGameDataStream = tEnv.toChangelogStream(flyGameTableResult).map(it ->
                new FlyingFlowerOrderData(it.<Integer>getFieldAs(0),
                        it.<Integer>getFieldAs(1), it.<Integer>getFieldAs(2), it.<Double>getFieldAs(3))
        );
        flyGameDataStream.print();
        SingleOutputStreamOperator<PoemRecommend> poemRecommendDataStream = tEnv.toChangelogStream(poemRecommendResult).map(it ->
                new PoemRecommend(it.<Integer>getFieldAs(0), it.<Integer>getFieldAs(1), it.<Integer>getFieldAs(2), System.currentTimeMillis())
        );
        poemRecommendDataStream.print();


        // 6. 存入数据库
        SinkFunction<FlyingFlowerOrderData> jdbcSinkFlyGame = JdbcSink.sink(
                "insert db_person.tb_fly_flower_order(uid, win_rate, win_rounds, total_rounds) values(?, ?, ?, ?)",
                (JdbcStatementBuilder<FlyingFlowerOrderData>) (preparedStatement, orderData) -> {
                    preparedStatement.setInt(1, orderData.getUid());
                    preparedStatement.setDouble(2, orderData.getWinRate());
                    preparedStatement.setInt(3, orderData.getWinRounds()); //
                    preparedStatement.setInt(4, orderData.getTotalRounds()); //
                },
                JdbcExecutionOptions.builder()
                        .withMaxRetries(3)
                        .withBatchSize(100)
                        .withBatchIntervalMs(3000)
                        .build()
                ,
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:mysql://localhost:3306/db_person?serverTimezone=Asia/Shanghai")
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        .withUsername("root")
                        .withPassword("123456")
                        .withConnectionCheckTimeoutSeconds(60) // 长连接
                        .build()
        );

        SinkFunction<PoemRecommend> jdbcSinkPoemRecommend = JdbcSink.sink(
                "insert db_person.tb_poem_recommend(uid, poem_id, click_count, time_stamp) values(?, ?, ?, ?)",
                (JdbcStatementBuilder<PoemRecommend>) (preparedStatement, poemRecommend) -> {
                    preparedStatement.setInt(1, poemRecommend.getUid());
                    preparedStatement.setDouble(2, poemRecommend.getPoemId());
                    preparedStatement.setInt(3, poemRecommend.getClickCount()); //
                    preparedStatement.setLong(4, poemRecommend.getTimeStamp()); //
                },
                JdbcExecutionOptions.builder()
                        .withMaxRetries(3)
                        .withBatchSize(100)
                        .withBatchIntervalMs(3000)
                        .build()
                ,
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:mysql://localhost:3306/db_person?serverTimezone=Asia/Shanghai")
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        .withUsername("root")
                        .withPassword("123456")
                        .withConnectionCheckTimeoutSeconds(60) // 长连接
                        .build()
        );

        flyGameDataStream.addSink(jdbcSinkFlyGame);
        poemRecommendDataStream.addSink(jdbcSinkPoemRecommend);

        // 7. 执行任务
        env.execute();
    }
}
